package com.atguigu.gmall.realtime.app.dim;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.app.function.CheckTableFunction;
import com.atguigu.gmall.realtime.app.function.DimSinkFunction;
import com.atguigu.gmall.realtime.app.function.TableProcessFunction;
import com.atguigu.gmall.realtime.bean.TableProcess;
import com.atguigu.gmall.realtime.util.MyKafkaUtil;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.source.MySqlSourceBuilder;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.streaming.api.datastream.BroadcastConnectedStream;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;

import java.io.Serializable;

public class DimApp {
    public static void main(String[] args) throws Exception {
        //TODO 1.基本环境准备
        //1.1 指定流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //1.2 设置并行度
        env.setParallelism(4);

        /*
        //TODO 2.检查点相关设置
        //2.1 开启检查点
        env.enableCheckpointing(5000L);
        //2.2 设置检查点超时时间
        env.getCheckpointConfig().setCheckpointTimeout(60000L);
        //2.3 设置两个检查点之间最小时间间隔
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(2000L);
        //2.4 job取消后，检查点是否保留
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        //2.5 设置重启策略
        // env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3,3000L));
        env.setRestartStrategy(RestartStrategies.failureRateRestart(3, Time.days(30),Time.seconds(3)));
        //2.6 设置状态后端
        env.setStateBackend(new HashMapStateBackend());
        // env.getCheckpointConfig().setCheckpointStorage(new JobManagerCheckpointStorage());
        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop202:8020/gmall/ck");
        //2.7 设置操作hdfs的用户
        System.setProperty("HADOOP_USER_NAME","atguigu");
        */

        //TODO 3.主流接收topic_db ( 启动maxwell)
        //3.1 声明消费的主题以及消费者组
        String topic = "topic_db";
        String groupId = "dim_app";
        //3.2 创建消费者对象
        FlinkKafkaConsumer<String> kafkaConsumer = MyKafkaUtil.getKafkaConsumer(topic, groupId);
        //3.3 消费数据 封装为流
        DataStreamSource<String> kafkaStream = env.addSource(kafkaConsumer);

        //TODO 4.主流转换结构 jsonStr->jsonObject
        SingleOutputStreamOperator<JSONObject> jsonObjStream = kafkaStream.map(JSON::parseObject);
        // jsonObjStream.print();

        //TODO 5.定义cdc数据源（ mysql配置表要建好，binlog要可以采集）
        MySqlSource<String> source = MySqlSource.<String>builder()
                .hostname("hadoop202")
                .port(3306)
                .databaseList("gmall0620_config") // set captured database
                .tableList("gmall0620_config.table_process") // set captured table
                .username("root")
                .password("123456")
                .deserializer(new JsonDebeziumDeserializationSchema()) // converts SourceRecord to JSON String
                .build();
        //TODO 6. cdc数据源加载为数据流（配置流）
        DataStreamSource<String> cdcStream = env.fromSource(source, WatermarkStrategy.noWatermarks(), "cdc_source");
        ///DataStreamSource<String> cdcStream = env.addSource(source);

        //TODO 7.提前创建维度表
        // 此处 cdcStream 的并行度为1
        // 1  如果phoenix没有表 创建   2如果配置表字段增加 ， phoenix也增加字段
        SingleOutputStreamOperator<String> cdcCheckedStream = cdcStream.map(new CheckTableFunction());


        //TODO 7.要把配置流的数据广播进主流  两流合一流
        //7.1 对维度配置状态的定义
        MapStateDescriptor<String, TableProcess> tableProcessStateDesc = new MapStateDescriptor<String, TableProcess>("table_process", String.class, TableProcess.class);
        //7.2 把配置流变为广播流
        BroadcastStream<String> broadcastStream = cdcCheckedStream.broadcast(tableProcessStateDesc);
        //7.3 把主流与广播流合流
        BroadcastConnectedStream<JSONObject, String> dataWithBroadCastStream = jsonObjStream.connect(broadcastStream);


        //TODO 8.实现一个方法处理数据（1 处理数据变化  2 处理配置变化）
        SingleOutputStreamOperator<JSONObject> dataForSinkStream = dataWithBroadCastStream.process(new TableProcessFunction(tableProcessStateDesc));

        //  测试：1  事实表是否被过滤掉 2 字段是否经过剪裁 3 是否添加了 最终要保存的目标表标识
        dataForSinkStream.print("最终要保存到维度的数据：");

        //TODO 9.保存到phoenix
        dataForSinkStream.addSink(new DimSinkFunction());

        env.execute();

    }
}
